package com.shujia.flink.tf;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Demo2FlatMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.readTextFile("flink/data/words.txt");

        //匿名内部类方式
        DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {
            /**
             * flatMap方法每一条数据执行一次，进来一条出去多条
             * @param value 一行数据
             * @param out 用于将数据发送到下游
             */
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] split = value.split(",");
                for (String word : split) {
                    //将数据发送到下游
                    out.collect(word);
                }
            }
        });
        //wordsDS.print();
        //lambda表达式
        DataStream<String> wordsDS2 = linesDS
                .flatMap((line, out) -> {
                    for (String word : line.split(",")) {
                        out.collect(word);
                    }
                }, Types.STRING);

        wordsDS2.print();


        env.execute();
    }
}
